package com.gao.flying.mq;

import com.gao.flying.mq.annotation.Consumer;
import com.gao.flying.mq.config.Broker;
import com.gao.flying.mq.config.Dest;
import com.gao.flying.mq.config.Scan;
import com.gao.flying.mq.consumer.JmsConsumer;
import com.gao.flying.mq.consumer.JmsDurableConsumer;
import com.gao.flying.mq.core.JmsContext;
import com.gao.flying.mq.core.SessionCallback;
import com.gao.flying.mq.support.ClassUtil;
import com.gao.flying.mq.util.XmlUtils;
import org.apache.commons.lang3.StringUtils;

import javax.jms.*;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class JmsClient {

    private static Map<String, JmsContext> contextMap = new ConcurrentHashMap<>();
    private static Lock lock = new ReentrantLock();
    private static volatile Boolean startedConsumer = Boolean.FALSE;
    private static ExecutorService executor = Executors.newCachedThreadPool();

    private JmsClient() {
    }

    public static JmsContext getContext(String destinationName) {
        if (!contextMap.containsKey(destinationName)) {
            lock.lock();
            try {
                Dest dest = XmlUtils.getDestination(destinationName);
                if (dest == null || dest.getName().isEmpty()) {
                    throw new RuntimeException("no such queue :" + destinationName);
                }

                Broker broker = XmlUtils.getBroker(dest.getBroker());
                if (broker == null || broker.getName().isEmpty()) {
                    throw new RuntimeException("no such broker :" + dest.getBroker());
                }

                JmsContext context = new JmsContext(broker.getUrl());
                contextMap.put(destinationName, context);
            } finally {
                lock.unlock();
            }
        }
        return contextMap.get(destinationName);
    }


    public static JmsContext getContext(String destinationName, String brokerName) {
        if (!contextMap.containsKey(destinationName)) {
            lock.lock();
            try {
                Broker broker = XmlUtils.getBroker(brokerName);
                if (broker == null || broker.getName().isEmpty()) {
                    throw new RuntimeException("no such broker :" + brokerName);
                }
                JmsContext context = new JmsContext(broker.getUrl());
                contextMap.put(destinationName, context);
            } finally {
                lock.unlock();
            }
        }
        return contextMap.get(destinationName);
    }


    public static void sendMessage(String destinationName, Object message) throws JMSException {
        Dest dest = XmlUtils.getDestination(destinationName);
        if (dest == null) {
            throw new RuntimeException("no such destination:" + destinationName);
        }
        send(destinationName, dest.getType(), dest.getDurable(), message);
    }

    protected static <T> T execute(String destinationName, SessionCallback<T> action) throws JMSException {
        Session sessionToClose = getContext(destinationName).getSession();
        try {
            T t = action.doInJms(sessionToClose);
            return t;
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            if (sessionToClose != null) {
                sessionToClose.close();
            }
        }
        return null;
    }

    protected static void send(final String destinationName, String destinationType, Boolean durable, final Object message) throws JMSException {
        execute(destinationName, session -> {
            Destination destination = null;
            if (destinationType.equalsIgnoreCase("queue")) {
                destination = session.createQueue(destinationName);
            } else if (destinationType.equalsIgnoreCase("topic")) {
                destination = session.createTopic(destinationName);
            } else {
                throw new RuntimeException("error destination type:" + destinationType);
            }
            doSend(destinationName, durable, session, destination, message);
            return null;
        });
    }

    protected static void doSend(String destinationName, Boolean durable, Session session, Destination destination, Object message) throws JMSException {
        MessageProducer producer = session.createProducer(destination);
        if (durable) {
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        } else {
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        }
        Message msg = getContext(destinationName).toMessage(message, session);
        producer.send(msg);
    }

    public static void close(String destinationName) {
        JmsContext context = contextMap.get(destinationName);
        if (context != null) {
            context.close();
            contextMap.remove(destinationName);
        }
    }

    public static void closeAll() {
        for (String key : contextMap.keySet()) {
            contextMap.get(key).close();
        }
        contextMap.clear();
    }

    public static void startConsumer() {

        if (!startedConsumer) {
            Scan scan = XmlUtils.getScan();
            if (scan == null || StringUtils.isBlank(scan.getPath())) {
                throw new RuntimeException("please set scan path in jms.xml");
            }
            synchronized (startedConsumer) {
                List<Class<?>> classList = ClassUtil.getClasses(scan.getPath());
                classList.stream().filter(clazz -> clazz.isAnnotationPresent(Consumer.class)).forEach(clazz -> {
                    Consumer consumer = clazz.getAnnotation(Consumer.class);

                    String destinationName = consumer.destinationName();
                    String destinationType = consumer.destinationType();
                    String brokerName = consumer.brokerName();
                    boolean durable = consumer.durable();
                    String clientId = consumer.clientId();
                    JmsContext jmsContext = getContext(destinationName, brokerName);

                    try {
                        MessageListener listener = (MessageListener) clazz.newInstance();
                        if ("topic".equalsIgnoreCase(destinationType) && durable) {
                            if (StringUtils.isBlank(clientId)) {
                                throw new RuntimeException("annotation Consumer:" + destinationName + " not set client");
                            }

                            JmsDurableConsumer jmsDurableConsumer = new JmsDurableConsumer(clientId, destinationName, jmsContext, listener);
                            executor.execute(jmsDurableConsumer);
                        } else {
                            JmsConsumer jmsConsumer = new JmsConsumer(destinationName, destinationType, jmsContext, listener);
                            executor.execute(jmsConsumer);
                        }

                    } catch (InstantiationException | IllegalAccessException e) {
                        e.printStackTrace();
                    }
                });

                startedConsumer = Boolean.TRUE;
            }
        } else {
            throw new RuntimeException("consumer listener is already started!");
        }
    }

}
